package com.ctyun.dws

import com.shujia.utils.{DateUtils, SparkTool}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.storage.StorageLevel

object DwsCompanyTimeSpaceMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._




    // 1、从MySQl中读取出 确诊名单
    val confirmMdnDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306/crm")
      .option("dbtable", "confirm")
      .option("user", "root")
      .option("password", "123456")
      .load()
      .select($"mdn")

    // 2、读取dwi层中的位置数据融合表
    val mergeLocationDF: Dataset[Row] = spark
      .table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)
      .select($"mdn", $"start_date", $"longi", $"lati", $"county_id", $"county_id", $"end_date", $"bsid", $"grid_id")

    // 3、从位置数据融合表中找出确诊名单中这些人员 一天的行动轨迹
    val confirmLocationDF: DataFrame = mergeLocationDF
      .join(confirmMdnDF.hint("broadcast"), "mdn")
      .select($"mdn" as "c_mdn", $"start_date" as "c_start_date", $"longi" as "c_longi", $"lati" as "c_lati", $"county_id")

    // 4、基于位置数据以及确诊人员的行动轨迹找出时空伴随者
    /**
     * 时空伴随者的条件：
     * 1、时间：小于1个小时
     * 2、空间：小于500m
     */
    // 通过区县id进行关联 以每个区县分组 将位置数据进行对比
    val companyOfTimeAndSpaceDF: Dataset[Row] = mergeLocationDF
      .join(confirmLocationDF.hint("broadcast"), "county_id")
      // 过滤掉已经确诊的人员位置记录
      .where($"mdn" =!= $"c_mdn")
      // 计算时间
      .withColumn("diff_time", dateDiff($"start_date", $"c_start_date"))
      .where($"diff_time" < 3600)
      // 计算距离
      .withColumn("distance", calculateLength($"longi", $"lati", $"c_longi", $"c_lati"))
      .where($"distance" < 500)
      .select($"mdn"
        ,$"start_date"
        ,$"end_date"
        ,$"county_id"
        ,$"longi"
        ,$"lati"
        ,$"bsid"
        ,$"grid_id"
        ,$"c_mdn"
        ,$"c_start_date"
        ,$"c_longi"
        ,$"c_lati")



    // 对多次使用的DF进行缓存
    companyOfTimeAndSpaceDF.persist(StorageLevel.MEMORY_AND_DISK_SER)

    // 将数据保存
    companyOfTimeAndSpaceDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_company_time_space_msk_d/day_id=$day_id")

    // 增加分区
    spark.sql(
      s"""
         |alter table dws.dws_company_time_space_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)

    // 将时空伴随者的手机号单独保存
    companyOfTimeAndSpaceDF
      .select($"mdn")
      .distinct()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_company_time_space_mdn_msk_d/day_id=$day_id")

    // 增加分区
    spark.sql(
      s"""
         |alter table dws.dws_company_time_space_mdn_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)


  }
}
